
----------------------------------------------------------------
--- Copyright：  迷你玩-海马
--- Description：事件入队和出队列存储器 使用ngx共享内存队列来实现
---              提供类似 message-queue 的先进先出事件处理
---              核心是基于 TimeWheel 的检测队列
---
---   使用场景：  1 放入一个延迟事件，系统每N秒，批量执行M个
---
---   API: new   初始化：建立队列，并设置回调函数
---   API: push  增加一个事件，N秒后执行（可以插队，会返回当时的队列长度）
---
---
---   例子：
---
---		-- 事件的回调函数
---		function  FOO.func_cb( event )
---			echo 'my uin is ' .. event.uin
---		end
---
---		-- 初始化
---		local t_ = {
---			name='test',               --队列名字
---			share=ngx_shared.miniw,    --share_dict
---			cb=FOO.func_cb,            --回调函数
---			wheel_time = 5,            --检查事件时间间隔
---		}
---		FOO.my_event_mq = CEventMQ:new( t_ )
---		FOO.my_event_mq:start_timer_every()    --启动定时器

---
---     -- 加入一个事件， 延迟10秒执行(一般是重试一个事件)
---     FOO.my_event_mq:push( { uin=12345, check_dup=loop_12345 }, { delay=10 } )
---
-----------------------------------------------------------------


local assert     = assert
local require    = require    
local bit        = bit
local coroutine  = coroutine
local debug      = debug
local io         = io
local pairs      = pairs
local ipairs     = ipairs
local math       = math
local os         = os
local print      = print
local pcall      = pcall
local xpcall     = xpcall
local rawget     = rawget
local rawset     = rawset
local select     = select
local string     = string
local table      = table
local tonumber   = tonumber
local tostring	 = tostring
local error      = error
local type       = type
local unpack     = unpack
local setmetatable = setmetatable
local getmetatable = getmetatable
local ngx = ngx

local math_floor = math.floor
local table_tostring   = table.tostring
local table_loadstring = table.loadstring


----数据变量----
local uu         = require("modules.commhm.utils");
local log        = require("modules.commhm.log");
local ns_time    = require("modules.commhm.time");
--local CNgxTimerWrap   = require("modules.commhm.ngxTimerWrap" )

-- 事件类  post_event cb_event
local _M = { _VERSION = '0.12.1' }
local mt = { __index = _M }



--[[
-------- 数据结构:

TimeWheel0 = push_pop队列 { 1,2,,6 }
TimeWheel1 = push_pop队列 { 1,2,3,4,5,6 }
TimeWheel2 = push_pop队列 { 1,2,3,4, }
TimeWheel3 = push_pop队列 { 1,2,3, }
TimeWheel4 = push_pop队列 { 1,2,6 }
TimeWheel5 = push_pop队列 { 1,2,5,6 }
TimeWheel6 = push_pop队列 { 1,2,3,6 }
TimeWheel7 = push_pop队列 { 1,2,3,6 }
TimeWheel8 = push_pop队列 { 1,2,3 }
TimeWheel9 = push_pop队列 { 1,2,5,7 }

如果  检查间隔是6秒， 则最大支持延迟=6*10=60秒，
     检查间隔是10秒，则最大延迟时间=10*10=100秒

     定时器轮训扫描所有的TimeWheelX
--]]




---- 建立一个事件存储器
function _M:new( pp_ )
	log.debug( log.fast() and "call CEventMQ:new" .. (pp_.name or nil) )
	if  not pp_.share then
		log.error( 'CEventMQ:new share is nil' )
	end

	local ins_ = {
		name  = pp_.name,        -- 队列名字
		share = pp_.share,       -- 共享内存
		cb    = pp_.cb,          -- 回调函数

		wheel_count = pp_.wheel_count or 10,    -- 时间轮单点个数
		wheel_time  = pp_.wheel_time  or 1,     -- 检查间隔   (事件最大延迟为 wheel_count*wheel_time)
		max_count   = pp_.max_count   or 256,   -- 本时间轮单点 最大个事件个数


		---- 私有变量
		index        = 0,  -- 当前在用的节点
	}

	ins_.dup_ttl = ins_.wheel_count * ins_.wheel_time   --排重key的ttl

	return setmetatable(ins_, mt)
end



---- 插入一个事件 event是一个lua-table
--- event : 事件table内容
		-- check_dup  -- 是否排重，有值表示需要按这个值判断重复  排重时间ttl=wheel_count * wheel_time
--- pps   : 其他参数列表
		-- insert=1   -- 本时间轮的节点里优先插队
		-- delay=6    -- 延迟多少秒执行  ( wheel_time的整数倍  6 12 18 24 )
---- 返回值1:  元素个数， 0=已经存在
---- 返回值2:  err错误信息
function _M:push( event_, pps_ )

	--- 排重
	if  not pps_ then pps_ = {}	end
	if  event_.check_dup then
		if  self.share:get( event_.check_dup ) then
			log.day_list( 'CEventMQ', ngx.worker.id(), self.name, 'dup_existed', event_.check_dup )
			return 0, 'dup_existed'   -- 已经存在
		end
	end


	--- 计算加入的节点，至少下一个节点
	local add_ = 1
	if  pps_.delay then
		add_ = math_floor(pps_.delay / self.wheel_time)   ---5/5  10/5
		if  add_ <= 0 then
			add_ = 1                         --至少下一个节点
		elseif add_ >= self.wheel_count then
			add_ = self.wheel_count - 1      --至多最后一个节点
		end
	end


	local tw_name_ = self:get_wheel_name(add_)             --my_name3

	local len_ = self.share:llen(tw_name_)
	if  len_ >= self.max_count then
		log.error( 'error_full' )
		return 0, 'error_full'
	end

	local str_ = table_tostring( event_ )
	local count_, err_

	if  pps_.insert == 1 then
		count_, err_ = self.share:rpush( tw_name_, str_ )    --最右插队
	else
		count_, err_ = self.share:lpush( tw_name_, str_ )	 --从左顺序插入
	end


	if  event_.check_dup then
		self.share:set( event_.check_dup, 1, self.dup_ttl )   --设置排重
		--log.day_list( 'CEventMQ', ngx.worker.id(), self.name, 'set_dup', event_.check_dup )
	end

	log.day_list( 'CEventMQ', ngx.worker.id(), self.name, 'push', tw_name_, count_, (err_ or '') )
	return count_, err_
end



--- 获得时间轮节点
--- add_ = 0 当前节点
--- add_ > 0 下N个节点
function _M:get_wheel_name( add_ )
	if  add_ > 0 then
		return self.name .. ( (self.index + add_) % self.wheel_count )   ---0 to 9
	else
		return self.name .. self.index
	end
end



--- 处理一次扫描，将时间轮中的触发节点内容全部处理完
function _M:wheel_time_handle()
	local tw_name_ = self:get_wheel_name(0)       --my_name3
	self.index = self.index + 1
	if  self.index >= self.wheel_count then
		self.index = 0
	end

	local start_ = ngx.now()

	local len_ = self.share:llen(tw_name_)
	if  len_ and len_ > 0 then
		for i=1, self.max_count do
			local str_, err_ = self.share:rpop( tw_name_ )
			if  err_ then
				log.day_list( 'CEventMQ', ngx.worker.id(), self.name, 'error_pop', tw_name_, (err_ or '') )
				log.error( 'error_pop_' .. tw_name_ .. '_' .. (err_ or '') )
			end

			if  str_ then
				log.day_list( 'CEventMQ', ngx.worker.id(), self.name, 'pop', tw_name_, #str_ )
				local event_ = table_loadstring( str_ )
				if  event_ then
					if  event_.check_dup then
						self.share:delete( event_.check_dup )   --去掉排重标志
						--log.day_list( 'CEventMQ', ngx.worker.id(), self.name, 'delete_dup', event_.check_dup )
					end
					self.cb( event_ )
				end
			else
				break
			end

		end

		local cost_ = math_floor( (ngx.now() - start_ )*1000 + 0.5 );
		log.day_list( 'CEventMQ', ngx.worker.id(), self.name, 'pop_handle_cost', tw_name_, len_, cost_ )
	end

end



--- 开启定时器
function _M:start_timer_every()
	--- timer间隔，每N秒检查一下
	if  self.has_start_timer then
		-- 定时已经开启
		log.error( 'start_timer_every again' )
	else
		self.has_start_timer = 1

		--self.m_params = ngx.ctx.m_params or {}  --保存上下文
		--if  not self.m_params.uin then
			--self.m_params.uin = self.uin
		--end

		local function func_(premature)
			if  not  ngx.ctx.m_params then
				ngx.ctx.timer_at = 1
				ngx.ctx.log      = 0
				--ngx.ctx.m_params = self.m_params   --不加稍微提高一点性能
				ngx.ctx.xff      = "timer"
			end
			ngx.ctx.timer_start_time = ngx.now()
			ngx.ctx.premature = premature
			self:wheel_time_handle()
		end

		ngx.timer.every( self.wheel_time, func_ )
	end
end



return _M;
